package nsq

/*
*nsq生产者
 */

import (
	"log"
	"os"
	"time"

	"github.com/nsqio/go-nsq"
)

type Producer struct {
	producer *nsq.Producer
	nsqdAddr []string
	pos      int
	len      int
}

/*
*初始化生产者
*参数说明:
*@param:nsqdAddr nsqd 集群
 */
func NewProducer(nsqdAddr []string) (*Producer, error) {
	cfg := nsq.NewConfig()
	var err error
	producer := &Producer{nsqdAddr: nsqdAddr}
	producer.len = len(nsqdAddr)
	producer.pos = 0
	producer.producer, err = nsq.NewProducer(nsqdAddr[producer.pos], cfg)
	producer.producer.SetLogger(log.New(os.Stderr, "", log.Flags()), nsq.LogLevelError)
	if err != nil || producer.producer == nil {
		log.Fatal(err)
		return nil, err
	}
	return producer, nil
}

/*
*生产者nsq
*参数说明:
*@param:nsqdAddr		nsqd地址
*@param:topic			话题
 */
func (self *Producer) Publish(topic string, msg []byte) error {
	for err := self.producer.Publish(topic, msg); err != nil; err = self.producer.Publish(topic, msg) {
		//切换IP重连
		var err error
		self.producer.Stop()
		cfg := nsq.NewConfig()
		if self.pos >= self.len {
			self.pos = 0
		} else {
			self.pos++
		}
		self.producer, err = nsq.NewProducer(self.nsqdAddr[self.pos], cfg)
		self.producer.SetLogger(log.New(os.Stderr, "", log.Flags()), nsq.LogLevelError)
		if err != nil {
			log.Fatal(err)
		}
		//1秒后重新链接
		time.Sleep(time.Second)
	}
	return nil
}

/*
*关闭生产者
 */
func (self *Producer) Close() error {
	err := self.producer.Ping()
	if nil != err {
		self.producer.Stop()
		self.producer = nil
		return err
	}
	return nil
}
